package cn.dmrliu.edu.realtime.util;

import cn.dmrliu.edu.realtime.common.EduConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

public class DimUtil {

    /**
     * 查找维度信息
     *  1. 先从redis中查询，查到就返回
     *  2. 查不到再去phoenix中查询，并缓存到redis
     * @param tableName
     * @param keysTuple2
     * @return
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... keysTuple2) {

        // 拼接phoenix查询语句 和 redis的key
        StringBuilder selectSql = new StringBuilder();
        selectSql.append("select * from ").append(EduConfig.PHOENIX_SCHEMA).append(".").append(tableName.toUpperCase()).append(" where ");

        StringBuilder redisKey = new StringBuilder();
        redisKey.append("dim:").append(tableName.toLowerCase()).append(":");
        for (int i = 0; i < keysTuple2.length; i++) {
            Tuple2<String, String> keyEntry = keysTuple2[i];
            String columnName = keyEntry.f0;
            String columnValue = keyEntry.f1;

            selectSql.append(columnName).append("='").append(columnValue).append("'");
            redisKey.append(columnValue);

            if (i < keysTuple2.length - 1) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        // 从redis中获取值
        Jedis jedis = null;
        String dimJsonStr = null;
        JSONObject dimJsonObject = null;
        try {
            jedis = RedisUtil.getConnection();
            dimJsonStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 如果获取到直接返回
        if (StringUtils.isNotEmpty(dimJsonStr)) {
            dimJsonObject = JSON.parseObject(dimJsonStr);
        } else {
            // 如果获取不到，从phoenix中获取，并保存到redis中
            System.out.println("从phoenix表中查询维度：" + selectSql);

            List<JSONObject> list = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (list != null && list.size() == 1) {
                dimJsonObject = list.get(0);
                if (jedis != null) {
                    // 保存到redis，设置1天保存期
                    jedis.setex(redisKey.toString(), 3600 * 24, dimJsonObject.toJSONString());
                }
            } else {
                System.out.println("没有查找维度" + redisKey);
            }
        }
        if (jedis != null) {
            jedis.close();
        }

        return dimJsonObject;
    }

    // 用表名和key 获取 维度信息
    public static JSONObject getDimInfo(String tableName, String key) {
        return getDimInfo(tableName, Tuple2.of("id", key));
    }
}
